package com.rocketmq.simple;

import com.rocketmq.JmsProperties;
import com.rocketmq.MessageDto;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;

/**
 * 异步发送
 * @author: hw
 * @date: 2021/3/24 14:40
 */
@Slf4j
@Configuration
public class AsyncSend {


    @Autowired
    private JmsProperties jmsProperties;

    @Autowired
    private AsyncSendCallback asyncSendCallback;


    public void asyncSend(MessageDto messageDto) throws InterruptedException {
        // 消息创建者组
        DefaultMQProducer producer = new DefaultMQProducer(jmsProperties.getAsyncProducerGroup());
        // 添加服务器地址
        producer.setNamesrvAddr(jmsProperties.getServerAddr());
        final CountDownLatch start = new CountDownLatch(1);
        try {
            producer.setInstanceName(UUID.randomUUID().toString());
            // 启动Producer实例 bean格式已经启动过不需要手动启动
            producer.start();
            //发送异步失败的重试次 数
            producer.setRetryTimesWhenSendAsyncFailed(3);
            Message msg = new Message(messageDto.getTopic(), messageDto.getTag(), messageDto.getMessageBody().getBytes());
            producer.send(msg,asyncSendCallback);
        } catch (InterruptedException | MQClientException e) {
            e.printStackTrace();
           log.error("AyncSend类 43行 异常信息:{}",e.getMessage());
            Thread.currentThread().interrupt();
        } catch (RemotingException e) {
            throw new RuntimeException(e);
        } finally {
            start.countDown();
        }
        // 关闭Producer实例。
//        start.await(5, TimeUnit.SECONDS);
        producer.shutdown();
    }
}
